package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.CartAddUuBean;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.MyClickHouseUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author Felix
 * @date 2022/8/7
 * 交易域 加购独立用户数
 */
public class DwsTradeCartAddUuWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关设置
        /*env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(300L);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(100L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(15), Time.seconds(3)));
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop202/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");*/

        //TODO 3.从kafka主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_cart_add";
        String groupId = "dws_trade_cart_add_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);
        //{"id":"2454","user_id":"4","sku_id":"5","sku_num":"3","source_type":"2401","source_type_name":"用户查询","ts":"1659857176"}
        // kafkaStrDS.print("kafka>>>>");

        //TODO 4.对读取的数据进行类型的转换  jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObj = kafkaStrDS.map(JSON::parseObject);

        //TODO 5.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<JSONObject> watermarkDS = jsonObj.assignTimestampsAndWatermarks(
            WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<JSONObject>() {
                        @Override
                        public long extractTimestamp(JSONObject element, long recordTimestamp) {
                            return element.getLong("ts") * 1000L;
                        }
                    }
                )
        );
        //TODO 6.按照用户的id进行分组
        KeyedStream<JSONObject, String> keyDs = watermarkDS.keyBy(r -> r.getString("user_id"));
        //TODO 7.使用Flink的状态编程，判断是否加购独立用户
        SingleOutputStreamOperator<JSONObject> processDS = keyDs.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            private ValueState<String> lastAddcart ;

            @Override
            public void open(Configuration parameters) throws Exception {
                lastAddcart = getRuntimeContext().getState(
                    new ValueStateDescriptor<String>(
                        "lastAddcart", String.class
                    )
                );
            }

            @Override
            public void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {
                String lastAdddate = lastAddcart.value();
                Long ts = value.getLong("ts");
                String curDate = DateFormatUtil.toDate(ts);
                if (StringUtils.isNotEmpty(lastAdddate)) {
                    if (!lastAdddate.equals(curDate)) {
                        lastAddcart.update(curDate);
                        out.collect(value);
                    }
                } else {
                    lastAddcart.update(curDate);
                    out.collect(value);
                }

            }
        });

        //TODO 8.开窗
        AllWindowedStream<JSONObject, TimeWindow> windowDS = processDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));

        //TODO 9.聚合计算
        SingleOutputStreamOperator<CartAddUuBean> aggDS = windowDS.aggregate(
            new AggregateFunction<JSONObject, Long, Long>() {
                @Override
                public Long createAccumulator() {
                    return 0L;
                }

                @Override
                public Long add(JSONObject value, Long accumulator) {
                    return ++accumulator;
                }

                @Override
                public Long getResult(Long accumulator) {
                    return accumulator;
                }

                @Override
                public Long merge(Long a, Long b) {
                    return null;
                }
            }, new AllWindowFunction<Long, CartAddUuBean, TimeWindow>() {
                @Override
                public void apply(TimeWindow window, Iterable<Long> values, Collector<CartAddUuBean> out) throws Exception {
                    for (Long value : values) {
                        CartAddUuBean resBean = new CartAddUuBean(
                            DateFormatUtil.toYmdHms(window.getStart()),
                            DateFormatUtil.toYmdHms(window.getEnd()),
                            value,
                            System.currentTimeMillis()
                        );
                        out.collect(resBean);
                    }
                }
            }
        );
        //TODO 10.将聚合的结果写到Clickhouse中
        aggDS.print();
        aggDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_trade_cart_add_uu_window values(?,?,?,?)"));
        env.execute();
    }
}
